package org.example.service;

import cn.hutool.core.lang.Console;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;
import org.example.grpc.TestFirstServiceGrpc;
import org.example.model.TestProto;

import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author LinRuChang
 * @version 1.0
 * @date 2025/10/13
 * @since 1.8
 **/
public class TestFirstService {

    // 使用jackson进行转成 json字符串
    static ObjectMapper mapper = new ObjectMapper();

    static {
        mapper.findAndRegisterModules();
    }

    @SneakyThrows
    public TestProto.ResponseData helloWorld(TestProto.RequestData requestData) {
        // 1. 设置服务端地址信息
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9999)
                .usePlaintext()  // 屏蔽ssl证书，否则发送请求失败
                .build();

        // 2. 调用服务
        TestFirstServiceGrpc.TestFirstServiceBlockingStub testFirstServiceBlockingStub = TestFirstServiceGrpc.newBlockingStub(managedChannel);
        TestProto.ResponseData responseData = testFirstServiceBlockingStub.helloWorld(requestData);
        return responseData;
    }

    @SneakyThrows
    public Iterator<TestProto.ResponseData> helloWorldBySSE(TestProto.RequestData requestData) {
        // 1. 设置服务端地址信息
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9999)
                .usePlaintext()  // 屏蔽ssl证书，否则发送请求失败
                .build();

        // 2. 调用服务
        TestFirstServiceGrpc.TestFirstServiceBlockingStub testFirstServiceBlockingStub = TestFirstServiceGrpc.newBlockingStub(managedChannel);

        // 3. 由于上面是启动阻塞等待渠道，只有服务端返回了所有消息，应用端才释放走下一行代码
        Iterator<TestProto.ResponseData> responseDataIterator = testFirstServiceBlockingStub.helloWorldBySSE(requestData);

        // 3. 获取结果迭代器
        return responseDataIterator;
    }

    @SneakyThrows
    public CountDownLatch helloWorldBySSEReal(TestProto.RequestData requestData) {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // 1. 设置服务端地址信息
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9999)
                .usePlaintext()  // 屏蔽ssl证书，否则发送请求失败
                .build();

        // 2. 调用服务
        TestFirstServiceGrpc.TestFirstServiceStub testFirstServiceStub = TestFirstServiceGrpc.newStub(managedChannel);

        // 3. 实时接收处理服务端推送过的消息
        AtomicInteger messageCount = new AtomicInteger();
        testFirstServiceStub.helloWorldBySSE(requestData, new StreamObserver() {

            @Override
            @SneakyThrows
            public void onNext(Object value) {
                // 处理消息
                Console.log("【处理】结果到消息{}：{}", messageCount.incrementAndGet(), mapper.writeValueAsString(value));
            }

            @Override
            public void onError(Throwable e) {
                // 处理错误消息
                Console.log("【异常】结果到消息{}", messageCount.incrementAndGet(), e);

                // 异常，强制结束主线程等待
                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                // 服务端渠道主动关闭
                Console.log("【结束】结果到消息渠道结束");

                // 服务端已推送完毕，强制结束主线程等待
                countDownLatch.countDown();
            }
        });


        return countDownLatch;
    }


    /**
     * 客户端流式RPC
     *
     * @return 待发送的请求消息实体
     */
    public StreamObserver<TestProto.RequestData> helloWorldByClientStreamRpc() {

        // 1. 设置服务端地址信息
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9999)
                .usePlaintext()  // 屏蔽ssl证书，否则发送请求失败
                .build();

        // 2. 调用服务
        TestFirstServiceGrpc.TestFirstServiceStub testFirstServiceStub = TestFirstServiceGrpc.newStub(managedChannel);

        StreamObserver<TestProto.RequestData> requestDataStreamObserver = testFirstServiceStub.helloWorldByClientStreamRpc(new StreamObserver<org.example.model.TestProto.ResponseData>() {
            // 处理服务端消息响应
            @Override
            @SneakyThrows
            public void onNext(TestProto.ResponseData value) {
                Console.log("【处理】结果到服务器响应消息：{}", mapper.writeValueAsString(value));
            }

            @Override
            public void onError(Throwable t) {
                Console.log("【处理】结果到服务器响应异常");
            }

            @Override
            public void onCompleted() {
                Console.log("【处理】结果到服务器响应完毕");
            }
        });


        return requestDataStreamObserver;
    }


    /**
     * 双向流式RPC
     *
     * @return 待发送的请求消息实体
     */
    public StreamObserver<TestProto.RequestData> helloWorldByWebSocket() {

        // 1. 设置服务端地址信息
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9999)
                .usePlaintext()  // 屏蔽ssl证书，否则发送请求失败
                .build();

        // 2. 调用服务
        TestFirstServiceGrpc.TestFirstServiceStub testFirstServiceStub = TestFirstServiceGrpc.newStub(managedChannel);

        AtomicInteger responseMessageCount = new AtomicInteger();
        StreamObserver<TestProto.RequestData> requestDataStreamObserver = testFirstServiceStub.helloWorldByWebSocket(new StreamObserver<org.example.model.TestProto.ResponseData>() {
            // 处理服务端消息响应
            @Override
            @SneakyThrows
            public void onNext(TestProto.ResponseData value) {
                Console.log("【helloWorldByWebSocket】【处理】结果到服务器响应消息【{}】：{}", responseMessageCount.incrementAndGet(), mapper.writeValueAsString(value));
            }

            @Override
            public void onError(Throwable t) {
                Console.log("【helloWorldByWebSocket】【处理】结果到服务器响应异常");
            }

            @Override
            public void onCompleted() {
                Console.log("【helloWorldByWebSocket】【处理】结果到服务器响应完毕 == 消息总数【{}】", responseMessageCount.get());
            }
        });


        return requestDataStreamObserver;
    }
}
